package my.thesis.bolts.fieldExtractorsReliable;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import my.thesis.analytics.metrics.ThroughputMetric;

import org.joda.time.DateTime;
import org.json.JSONArray;
import org.json.JSONObject;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class TweetFieldExtractorTopo1Reliable extends BaseRichBolt
{
	private static final long serialVersionUID = 1L;	
	private OutputCollector collector;
		
	transient ThroughputMetric _throughputMetric;
	private static Boolean metricsOn=false;

	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector)
	{
		this.collector=collector;
		
		if(metricsOn)
		{
			_throughputMetric=new ThroughputMetric();
			context.registerMetric("tupleCount", _throughputMetric, 30);
			_throughputMetric.setStartTime();
		}
	}
		
	public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) 
	{		
			outputFieldsDeclarer.declare( new Fields("id","millis","lang","hash"));
	}
	
	public void execute(Tuple tuple) 
	{
		if(metricsOn)
		{
			_throughputMetric.incrTuples();
		}
		
		Set<String> hashtags = new HashSet<String>();

		String jsonString = tuple.getStringByField("jsonString");
		
		JSONObject obj = null;
		try
		{
			obj = new JSONObject(jsonString);
		}
		catch(Exception ex)
		{
			System.out.println("Something went wrong...");
		}
		
		if(obj != null)
		{
			Long id = obj.getLong("id");
			String username=obj.getString("username");
			int followers = obj.getInt("followers");
			
			String lang = obj.getString("lang");
			
			String text = obj.getString("text");
			String country = obj.getString("country");
			
			//DateTime timestamp=new DateTime(obj.get("createdat"));
			//Long millis = timestamp.getMillis();
			
			
			Long millis = obj.getLong("millis");
			
			//TEST
			//DateTime timestamp = DateTime.now();
			//Long millis = timestamp.getMillis();
			//TEST
			
			Double  lat = obj.getDouble("lat");
			Double lon = obj.getDouble("lon");
			
			if(obj.has("hashtags"))
			{
				JSONArray hashtagsArray = obj.getJSONArray("hashtags");
				for (int i=0; i < hashtagsArray.length(); i++)
				{
					hashtags.add(hashtagsArray.getJSONObject(i).getString("text"));
				}
			}
							
			for (String hashtag : hashtags)
			{
				collector.emit(tuple,new Values(id,millis,lang,hashtag));
			}

			collector.ack(tuple);	
						
		}
		else
		{
			collector.fail(tuple);	
		}		
	}	
}
